package com.atguigu.edu.realtime220815.util;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Properties;

/**
 * @Classname KafkaUtils
 * @Description TODO
 * @Date 2023/2/15 18:27
 * @Created by lzx
 */
public class KafkaUtils {
    private static final String KAFKA_SERVERS = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    public static String getKafkaSqlWith(String topicName , String groupId){
        return "  WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = '"+topicName+"',\n" +
                " 'properties.bootstrap.servers' = '"+KAFKA_SERVERS+"',\n" +
                " 'properties.group.id' = '"+groupId+"',\n" +
                " 'scan.startup.mode' = 'group-offsets',\n" +
                " 'format' = 'json'\n" +
                "  )";
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic){
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVERS);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"");
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("default_topic", new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                return new ProducerRecord<>(topic,element.getBytes());
            }
        }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return kafkaProducer;
    }

    //获取FlinkKafkaConsumer
    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic,String groupId){
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVERS);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic, new DeserializationSchema<String>() {
            @Override
            public String deserialize(byte[] message) throws IOException {
                if(message!=null && message.length!=0){
                    return new String(message);
                }
                return null;
            }

            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return Types.STRING;
            }
        }, props);
        return kafkaConsumer;
    }

    public static String getTopicDbDDL(String groupId){
        return "create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`proc_time` as PROCTIME(),\n" +
                "`ts` string\n" +
                ")" + getKafkaDDL("topic_db",groupId);
    }

    public static String getKafkaDDL(String topic, String groupId) {

        return " with ('connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVERS + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'group-offsets')";
    }
    public static String getUpsertKafkaDDL(String topic) {

        return "WITH ( " +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVERS + "', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }
}
